//
// Created by TongHua on 2020/12/23.
//

#ifndef WEBSERVER_THREADPOOL_H
#define WEBSERVER_THREADPOOL_H

#include <list>
#include <stdio.h>
#include <exception>
#include <pthread.h>
#include "../SQL/sql_connection_pool.hpp"
#include "../Lock/Locker.hpp"

template < typename T >
class ThreadPool
{
public:
    ThreadPool();
    void Init(ConnectionPool* _conn ,int _thread_number = 8,int _max_request = 10000);
    ~ThreadPool();
    /**
     * 添加任务至线程池
     * @param _request
     * @return
     */
    bool Append(T* _request);

private:
    static void *Work(void* arg);            // 任务函数，负责从线程池中取出线程执行任务队列
    void Run();
private:
    int m_thread_num;                   // 线程数量
    int m_max_requests;                 // 最大任务队列数量
    pthread_t* m_threads;               // 线程池数组
    std::list< T* > m_workQueue;        // 任务队列
    Sem m_queueStat;                    // 信号量，标志是否有任务需要处理
    Locker m_queueLocker;               // 任务队列互斥锁
    bool m_stop;
    ConnectionPool* m_connPool;         // 数据库池
};

template < typename T >
ThreadPool<T>::ThreadPool():m_stop(false),m_threads(NULL),m_connPool(NULL) {

}

template < typename T >
void ThreadPool<T>::Init(ConnectionPool *_conn, int _thread_number, int _max_request) {
    if ( _thread_number <= 0 || _max_request <= 0 )
    {
        throw std::exception();
    }
    m_connPool = _conn;
    m_thread_num = _thread_number;
    m_max_requests = _max_request;
    m_threads = new pthread_t[m_thread_num];
    if ( !m_threads )
    {
        throw std::exception();
    }
    for(int i=0;i<m_thread_num;++i)
    {
        printf("create thread = [%d]\n",i);
        if ( pthread_create(m_threads + i, NULL, Work, this) != 0 )
        {
            delete[] m_threads;
            throw std::exception();
        }
        if ( pthread_detach(m_threads[i]) )
        {
            delete[] m_threads;
            throw std::exception();
        }
    }
}

template < typename T >
ThreadPool<T>::~ThreadPool() {
    delete[] m_threads;
    m_stop = true;
}

template < typename T >
bool ThreadPool<T>::Append(T *_request) {
    m_queueLocker.Lock();
    if ( m_workQueue.size() > m_max_requests )
    {
        m_queueLocker.UnLock();
        return false;
    }
    m_workQueue.push_back(_request);
    m_queueLocker.UnLock();
    m_queueStat.Post();                 // 唤醒wait的线程
    return true;
}

template < typename T >
void* ThreadPool<T>::Work(void *arg) {
    ThreadPool* pool = (ThreadPool*)arg;
    pool->Run();
    return pool;
}

template < typename T >
void ThreadPool<T>::Run() {
    while(!m_stop)
    {
        m_queueStat.Wait();             // 等待资源就绪
        m_queueLocker.Lock();
        if(m_workQueue.empty())
        {
            m_queueLocker.UnLock();
            continue;
        }
        T* request = m_workQueue.front();
        m_workQueue.pop_front();
        m_queueLocker.UnLock();
        if(!request)
        {
            continue;
        }
        ConnectionRALL connectionRall( &request->mysql, m_connPool );

        request->Process();
    }
}

#endif //WEBSERVER_THREADPOOL_H
